package com.atguigu.wordCount;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author gmd
 * @desc DataStream实现WordCount（任务槽共享）
 * @since 2024-11-24 09:34:18
 */
public class SlotShareGroup {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // 设置并行度为1
        env.setParallelism(1);

        // 读取socket的数据
        DataStreamSource<String> socketDS = env.socketTextStream("127.0.0.1", 7777);

        // TODO 3. 处理数据: 切换、转换、分组、聚合
        SingleOutputStreamOperator<Tuple2<String,Integer>> sum = socketDS
                .flatMap(
                        (String value, Collector<String> out) -> {
                            String[] words = value.split(" ");
                            for (String word : words) {
                                out.collect(word);
                            }
                        }
                )
                .returns(Types.STRING)
                .map(word -> Tuple2.of(word, 1))
                .slotSharingGroup("group001")
                .returns(Types.TUPLE(Types.STRING,Types.INT))
                .keyBy(value -> value.f0)
                .sum(1);

        // 输出
        sum.print();
        // 执行
        env.execute();
    }
}

/*
 * Slot特点和slot数量与并行度的关系
 * =====================================
 *
 * 1. Slot特点
 * ------------
 *
 * 1）均分隔离内存，不隔离cpu
 * 2）可以共享：
 *      同一个job中，不同算子的子任务 才可以共享 同一个slot，同时在运行的
 *      前提是，属于同一个 slot共享组，默认都是“default”
 *
 * 2. Slot数量与并行度的关系
 * -------------------------
 *
 * 1）slot是一种静态的概念，表示最大的并发上限
 *      并行度是一种动态的概念，表示 实际运行 占用了 几个
 *
 * 2）要求： slot数量 >= job并行度（算子最大并行度），job才能运行
 *    注意：如果是yarn模式，动态申请。申请的TaskManager数量 = job并行度 / 每个TM的slot数，向上取整
 *      比如session： 一开始 0个TaskManager，0个slot
 *          --》 提交一个job，并行度10
 *              --》 10/3,向上取整，申请4个TaskManager，
 *              --》 使用10个slot，剩余2个slot
 */